package com.gome.ocean.service.task;

import java.io.File;
import java.io.IOException;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.apache.commons.io.FileUtils;
import org.apache.commons.lang.StringUtils;
import org.quartz.SchedulerException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import com.alibaba.fastjson.JSON;
import com.gome.framework.util.SystemInfoUtil;
import com.gome.ocean.common.constants.DataxContants;
import com.gome.ocean.common.enums.State;
import com.gome.ocean.common.utils.TimeUtils;
import com.gome.ocean.dao.datax.JobDAO;
import com.gome.ocean.dao.datax.JobHistoryDAO;
import com.gome.ocean.dao.model.datax.JobBO;
import com.gome.ocean.dao.model.datax.JobHistoryBO;
import com.gome.ocean.service.scheduler.SchedulerServiceBean;
import com.gome.ocean.service.thread.ThreadFactory;

/**
 * 检查已完成任务 没有更新状态及任务详情的
 * 
 * @author haqiaolong
 */
@Component
public class CheckSuccessJobTask implements InitializingBean, DisposableBean {

    private static final Logger LOG = LoggerFactory.getLogger(CheckSuccessJobTask.class);

    private final ScheduledExecutorService schedule = Executors.newScheduledThreadPool(1);

    @Autowired
    private JobDAO               jobDAO;
    @Autowired
    private JobHistoryDAO        jobHistoryDAO;
    @Autowired
    private ThreadFactory        threadFactory;
    @Autowired
    private SchedulerServiceBean schedulerServiceBean;

    @Override
    public void afterPropertiesSet() throws Exception {

        schedule.scheduleAtFixedRate(new Runnable() {

            @Override
            public void run() {
                String port = SystemInfoUtil.getServerPort();
                String ip = SystemInfoUtil.getLocalAddress();
                List<JobHistoryBO> jobHistoryBOs = jobHistoryDAO.findAllJobListByIpAndState(ip.concat(":").concat(port),
                    State.RUN.getValue());
                if (jobHistoryBOs != null && !jobHistoryBOs.isEmpty()) {
                    for (JobHistoryBO jobHistoryBO : jobHistoryBOs) {
                        JobBO jobBO = jobDAO.findJobById(jobHistoryBO.getTJobId());
                        if (jobBO != null) {
                            // 判断当前任务是否运行
                            boolean exist = false;
                            if (jobBO.isCronJob()) {
                                try {
                                    exist = schedulerServiceBean.checkExists(jobBO);
                                } catch (SchedulerException e) {
                                    continue;
                                }
                            } else {
                                exist = threadFactory.exist(jobBO.getId());
                            }
                            if (!exist && State.RUN == State.getState(jobBO.getState())
                                || State.SUCCESS == State.getState(jobBO.getState())) {
                                // 如果不存在并且状态是正在运行的
                                if (StringUtils.isNotEmpty(jobHistoryBO.getLogUrl())) {
                                    File logFile = new File(jobHistoryBO.getLogUrl());
                                    if (!logFile.exists()) {
                                        jobBO.setState(State.FAIL.getValue());
                                        jobHistoryBO.setState(State.FAIL.getValue());
                                        jobHistoryBO.setErrorMessage("运行日志丢失！忽略！");
                                        jobHistoryBO.setStartTime(jobBO.getStartTime());
                                        jobHistoryBO.setEndTime(jobBO.getEndTime());
                                        jobHistoryBO.setSpeedBytes("0");
                                        jobHistoryBO.setSpeedRecords(0l);
                                        jobHistoryBO.setTotalRecords(0l);
                                        jobHistoryBO.setErrorRecords(0l);

                                        jobDAO.updateJobBO(jobBO);

                                        jobHistoryBO.setSubmitUser(jobBO.getSubmitUser());
                                        jobHistoryBO.setSubmitTime(jobBO.getSubmitTime());
                                        jobHistoryBO.setSubmitIp(jobBO.getSubmitIp());
                                        jobHistoryBO.setPipelineName(jobBO.getPipelineName());
                                        jobHistoryBO.setState(jobBO.getState());
                                        jobHistoryBO.setCronExpression(jobBO.getCronExpression());
                                        jobHistoryDAO.updateJobHistoryById(jobHistoryBO);
                                        LOG.info("插入历史记录完成！jobId:" + jobBO.getId());
                                    } else {
                                        try {
                                            String content = FileUtils.readFileToString(logFile, "utf-8");
                                            if (StringUtils.isNotEmpty(content)) {
                                                if (content.contains(DataxContants.DEFAULT_DATAX_ERROR)) {
                                                    jobBO.setState(State.FAIL.getValue());

                                                    jobHistoryBO.setState(State.FAIL.getValue());
                                                    jobHistoryBO.setErrorMessage(StringUtils.substringAfterLast(content,
                                                        DataxContants.DEFAULT_DATAX_ERROR));
                                                } else {
                                                    jobBO.setState(State.SUCCESS.getValue());
                                                    jobHistoryBO.setState(State.SUCCESS.getValue());
                                                }
                                                if (content.contains("logStatistics info:")) {
                                                    String subStr = StringUtils.substringAfterLast(content,
                                                        "logStatistics info:");
                                                    if (StringUtils.isNotEmpty(subStr)) {
                                                        subStr = StringUtils.substring(subStr,
                                                            subStr.indexOf("{"),
                                                            subStr.indexOf("}") + 1);
                                                        @SuppressWarnings("unchecked")
                                                        Map<String, Object> maps = JSON.parseObject(subStr, Map.class);
                                                        if (maps.containsKey("startTime")) {
                                                            jobHistoryBO.setStartTime(TimeUtils
                                                                .getDateTime(maps.get("startTime").toString()));
                                                        }
                                                        if (maps.containsKey("endTime")) {
                                                            jobHistoryBO.setEndTime(
                                                                TimeUtils.getDateTime(maps.get("endTime").toString()));
                                                            jobBO.setEndTime(jobHistoryBO.getEndTime());
                                                        }
                                                        if (maps.containsKey("byteSpeedPerSecond")) {
                                                            jobHistoryBO.setSpeedBytes(
                                                                maps.get("byteSpeedPerSecond").toString());
                                                        }
                                                        if (maps.containsKey("recordSpeedPerSecond")) { // 记录写入速度
                                                            jobHistoryBO.setSpeedRecords(Long
                                                                .valueOf(maps.get("recordSpeedPerSecond").toString()));
                                                        }
                                                        if (maps.containsKey("totalReadRecords")) { // 读出记录总数
                                                            jobHistoryBO.setTotalRecords(
                                                                Long.valueOf(maps.get("totalReadRecords").toString()));
                                                        }
                                                        if (maps.containsKey("totalErrorRecords")) { // 错误数
                                                            jobHistoryBO.setErrorRecords(
                                                                Long.valueOf(maps.get("totalErrorRecords").toString()));
                                                        }
                                                    }
                                                } else {
                                                    jobBO.setEndTime(new Date());
                                                    jobHistoryBO.setStartTime(jobBO.getStartTime());
                                                    jobHistoryBO.setEndTime(jobBO.getEndTime());
                                                    jobHistoryBO.setSpeedBytes("0");
                                                    jobHistoryBO.setSpeedRecords(0l);
                                                    jobHistoryBO.setTotalRecords(0l);
                                                    jobHistoryBO.setErrorRecords(0l);
                                                }

                                                jobDAO.updateJobBO(jobBO);

                                                jobHistoryBO.setSubmitUser(jobBO.getSubmitUser());
                                                jobHistoryBO.setSubmitTime(jobBO.getSubmitTime());
                                                jobHistoryBO.setSubmitIp(jobBO.getSubmitIp());
                                                jobHistoryBO.setPipelineName(jobBO.getPipelineName());
                                                jobHistoryBO.setState(jobBO.getState());
                                                jobHistoryBO.setCronExpression(jobBO.getCronExpression());
                                                jobHistoryDAO.updateJobHistoryById(jobHistoryBO);
                                                LOG.info("插入历史记录完成！jobId:" + jobBO.getId());
                                            }
                                        } catch (IOException e) {
                                            LOG.error("历史任务文件读取失败！", e);
                                        }
                                    }
                                } else {
                                    jobBO.setEndTime(new Date());
                                    jobBO.setState(State.FAIL.getValue());
                                    jobHistoryBO.setState(State.FAIL.getValue());
                                    jobHistoryBO.setErrorMessage("运行日志丢失！忽略！");
                                    jobHistoryBO.setStartTime(jobBO.getStartTime());
                                    jobHistoryBO.setEndTime(jobBO.getEndTime());
                                    jobHistoryBO.setSpeedBytes("0");
                                    jobHistoryBO.setSpeedRecords(0l);
                                    jobHistoryBO.setTotalRecords(0l);
                                    jobHistoryBO.setErrorRecords(0l);

                                    jobDAO.updateJobBO(jobBO);

                                    jobHistoryBO.setSubmitUser(jobBO.getSubmitUser());
                                    jobHistoryBO.setSubmitTime(jobBO.getSubmitTime());
                                    jobHistoryBO.setSubmitIp(jobBO.getSubmitIp());
                                    jobHistoryBO.setPipelineName(jobBO.getPipelineName());
                                    jobHistoryBO.setState(jobBO.getState());
                                    jobHistoryBO.setCronExpression(jobBO.getCronExpression());
                                    jobHistoryDAO.updateJobHistoryById(jobHistoryBO);
                                    LOG.info("插入历史记录完成！jobId:" + jobBO.getId());
                                }
                            }
                        }
                    }
                }

            }
        }, 0, 60, TimeUnit.SECONDS);
    }

    @Override
    public void destroy() throws Exception {
        schedule.shutdown();
    }
}
